package io.cici.worker.core.tracker.task.heavy;

import com.google.common.base.Stopwatch;
import io.cici.common.RemoteConstant;
import io.cici.common.safeRunnable.SafeRunnable;
import io.cici.worker.common.constants.TaskStatus;
import io.cici.worker.persistence.TaskDO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 定时扫描数据库中的task（出于内存占用量考虑，每次最多获取100个），并将需要执行的任务派发出去
 */
@Slf4j
class HeavyTaskTracker$Dispatcher extends SafeRunnable {

    // 数据库查询限制，每次最多查询几个任务
    private static final int DB_QUERY_LIMIT = 100;

    private final HeavyTaskTracker heavyTaskTracker;

    public HeavyTaskTracker$Dispatcher(HeavyTaskTracker heavyTaskTracker) {
        this.heavyTaskTracker = heavyTaskTracker;
    }

    @Override
    public void run0() {

        if (heavyTaskTracker.finished.get()) {
            return;
        }

        Stopwatch stopwatch = Stopwatch.createStarted();

        // 1. 获取可以派发任务的 ProcessorTracker
        List<String> availablePtIps = heavyTaskTracker.processorTrackerStatusHolder.getAvailableProcessorTrackers();

        // 2. 没有可用 ProcessorTracker，本次不派发
        if (availablePtIps.isEmpty()) {
            log.debug("[TaskTracker-{}] no available ProcessorTracker now.", heavyTaskTracker.instanceId);
            return;
        }

        // 3. 避免大查询，分批派发任务
        long currentDispatchNum = 0;
        long maxDispatchNum = availablePtIps.size() * heavyTaskTracker.instanceInfo.getThreadConcurrency() * 2L;
        AtomicInteger index = new AtomicInteger(0);

        // 4. 循环查询数据库，获取需要派发的任务
        while (maxDispatchNum > currentDispatchNum) {

            int dbQueryLimit = Math.min(DB_QUERY_LIMIT, (int) maxDispatchNum);
            List<TaskDO> needDispatchTasks = heavyTaskTracker.taskPersistenceService.getTaskByStatus(heavyTaskTracker.instanceId, TaskStatus.WAITING_DISPATCH, dbQueryLimit);
            currentDispatchNum += needDispatchTasks.size();

            needDispatchTasks.forEach(task -> {
                // 获取 ProcessorTracker 地址，如果 Task 中自带了 Address，则使用该 Address
                String ptAddress = task.getAddress();
                if (StringUtils.isEmpty(ptAddress) || RemoteConstant.EMPTY_ADDRESS.equals(ptAddress)) {
                    ptAddress = availablePtIps.get(index.getAndIncrement() % availablePtIps.size());
                }
                heavyTaskTracker.dispatchTask(task, ptAddress);
            });

            // 数量不足 或 查询失败，则终止循环
            if (needDispatchTasks.size() < dbQueryLimit) {
                break;
            }
        }

        log.debug("[TaskTracker-{}] dispatched {} tasks,using time {}.", heavyTaskTracker.instanceId, currentDispatchNum, stopwatch.stop());
    }
}
